package cn.baker.rabbitmq.config;

import cn.baker.rabbitmq.mapper.TransMessageMapper;
import cn.baker.rabbitmq.sender.TransMessageSender;
import cn.baker.rabbitmq.service.TransMessageService;
import cn.baker.rabbitmq.service.TransMessageServiceImpl;
import cn.baker.rabbitmq.task.ResendTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.*;
import org.springframework.jdbc.core.JdbcTemplate;

import javax.annotation.Resource;

/**
 * org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration
 * @author yuanyu
 */
@Slf4j
@Configuration
@EnableConfigurationProperties(TransRabbitProperties.class)
public class TransRabbitConfig {

    @Bean("transMessageMapper")
    public TransMessageMapper transMessageMapper(JdbcTemplate jdbcTemplate) {
        return new TransMessageMapper(jdbcTemplate);
    }

    @Bean("transMessageService")
    public TransMessageService transMessageService(TransMessageMapper transMessageMapper, TransRabbitProperties transRabbitProperties) {
        return new TransMessageServiceImpl(transMessageMapper, transRabbitProperties);
    }

    @Bean("transMessageSender")
    public TransMessageSender transMessageSender(@Qualifier("transRabbitTemplate") RabbitTemplate rabbitTemplate, @Qualifier("transMessageService") TransMessageService transMessageService) {
        return new TransMessageSender(rabbitTemplate, transMessageService);
    }

    /**
     * 连接工厂
     */
    @Bean("transRabbitConnectionFactory")
    public ConnectionFactory connectionFactory(TransRabbitProperties properties) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(properties.getHost());
        connectionFactory.setPort(properties.getPort());
        connectionFactory.setPassword(properties.getPassword());
        connectionFactory.setUsername(properties.getUsername());
        connectionFactory.setVirtualHost(properties.getVhost());
        // 发送者确认
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 消息返回（消息无法路由）
        connectionFactory.setPublisherReturns(true);
        connectionFactory.createConnection();
        return connectionFactory;
    }

    /**
     * 消息监听
     * @param connectionFactory
     * @return
     */
    @Bean("transRabbitListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(@Qualifier("transRabbitConnectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 并发消费者
        factory.setConcurrentConsumers(3);
        // 最大的并发消费者
        factory.setMaxConcurrentConsumers(10);
        // 手动接收
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }

    /**
     * 处理队列交换机声明与绑定
     * @param connectionFactory
     * @return
     */
    @Bean("transRabbitAdmin")
    public RabbitAdmin rabbitAdmin(@Qualifier("transRabbitConnectionFactory") ConnectionFactory connectionFactory) {
        RabbitAdmin admin = new RabbitAdmin(connectionFactory);
        // 自动
        admin.setAutoStartup(true);
        return admin;
    }

    /**
     * 自定义RabbitTemplate
     * @param connectionFactory
     * @return
     */
    @Bean("transRabbitTemplate")
    public RabbitTemplate customRabbitTemplate(@Qualifier("transRabbitConnectionFactory") ConnectionFactory connectionFactory, @Qualifier("transMessageService") TransMessageService transMessageService) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 打开托管
        rabbitTemplate.setMandatory(true);
        // 发送者返回
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
                    log.info("@mqLog 消息投递回调, correlationData:[{}], ack:[{}], cause:[{}]", correlationData, ack, cause);
                    if (ack && null != correlationData) {
                        String messageId = correlationData.getId();
                        log.info("@mqLog 消息已正确投递到交换机, messageId:[{}]", messageId);
                        transMessageService.messageSendSuccess(messageId);
                    } else {
                        // 投递前以及把消息存表了，这里只是报错 cn.baker.rabbitmq.sender.TransMessageSender.convertAndSend
                        log.error("@mqLog 消息投递至交换机失败, correlationData:[{}], cause:[{}]", correlationData, cause);
                    }
                }
        );

        // 消息无法路由
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.error("@mqLog 消息无法路由, message:[{}], replyCode:[{}], replyText:[{}], exchange:[{}], routingKey:[{}]", message, replyCode, replyText, exchange, routingKey);
            transMessageService.messageSendReturn(
                    message.getMessageProperties().getMessageId(),
                    exchange,
                    routingKey,
                    new String(message.getBody())
            );
        });

        return rabbitTemplate;
    }


    /**
     * ---------------------------------------------------------------------------------------
     */

    @Bean("defaultRabbitConnectionFactory")
    @Primary
    public ConnectionFactory defaultConnectionFactory(TransRabbitProperties properties) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(properties.getHost());
        connectionFactory.setPort(properties.getPort());
        connectionFactory.setPassword(properties.getPassword());
        connectionFactory.setUsername(properties.getUsername());
        connectionFactory.setVirtualHost(properties.getVhost());
        connectionFactory.createConnection();
        return connectionFactory;
    }
    @Bean("defaultRabbitAdmin")
    @Primary
    public RabbitAdmin defaultRabbitAdmin(@Qualifier("defaultRabbitConnectionFactory") ConnectionFactory connectionFactory) {
        RabbitAdmin admin = new RabbitAdmin(connectionFactory);
        // 自动
        admin.setAutoStartup(true);
        return admin;
    }
    @Bean("defaultRabbitListenerContainerFactory")
    @Primary
    public RabbitListenerContainerFactory<SimpleMessageListenerContainer> defaultRabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        return factory;
    }
    @Bean("defaultRabbitTemplate")
    @Primary
    public RabbitTemplate defaultRabbitTemplate(@Qualifier("defaultRabbitConnectionFactory") ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }






}
